package com.pw.study.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

import static org.apache.flink.table.api.Expressions.$;

public class $04TableKafka {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //初始化格式
        Schema schema = new Schema()
                .field("id", DataTypes.STRING())
                .field("ts", DataTypes.BIGINT())
                .field("vc", DataTypes.INT());

        tEnv.connect(new Kafka()
                        .version("universal")
                        .property("bootstrap.servers", "hadoop162:9092")
                        .property("group.id", "kafka1")
                        .topic("sql")
                        .startFromLatest()
                ).withFormat(new Csv())
                .withSchema(schema).createTemporaryTable("hello");
        // 读文件
        //mcReadKafka(tEnv);
        // 从kafka临时表中读取数据
        Table tb = tEnv.from("hello").select($("id"),$("ts"), $("vc"));
        //写入kafka
        tEnv.connect(new Kafka()
                .version("universal")
                .version("universal")
                .property("bootstrap.servers", "hadoop162:9092")
                .topic("ss2")
                .sinkPartitionerRoundRobin())
                .withFormat(new Json())
                .withSchema(schema).createTemporaryTable("tmp");
        tb.executeInsert("tmp");



    }

    private static void mcReadKafka(StreamTableEnvironment tEnv) {

        tEnv.from("hello").execute().print();
    }

}
